package cn.taqu.es.config;

import cn.taqu.es.util.JsonUtil;
import cn.taqu.es.util.TypeReference;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpHost;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

@Configuration
@Slf4j
public class ESConfig {

    private static final String ES_DEFAULT_SCHEMA = "http";
    public static int CONNECT_TIMEOUT_MILLIS = 1000;
    public static int SOCKET_TIMEOUT_MILLIS = 30000;
    public static int CONNECTION_REQUEST_TIMEOUT_MILLIS = 1000;
    public static int MAX_CONN_PER_ROUTE = 10;
    public static int MAX_CONN_TOTAL = 30;
    private RestClientBuilder builder;
    private RestClient restClient;
    private RestHighLevelClient restHighLevelClient;

    @Value("${es.hostList}")
    private String esHostJson;


    public static List<Map<String, Object>> initEsHostList(String json) {
        List<Map<String, Object>> mapList = JsonUtil.toList(json, new TypeReference<List<Map<String, Object>>>() {
        }.getType());
        return mapList;
    }

    @Bean
    public HttpHost[] esHttpHost() {
        List<Map<String, Object>> esHostList = initEsHostList(esHostJson);
        return esHostList.stream().map(esHost -> {
            String host = esHost.get("ip").toString();
            Integer port = Integer.parseInt(esHost.get("port").toString());
            return new HttpHost(host, port, ES_DEFAULT_SCHEMA);
        }).collect(Collectors.toList()).toArray(new HttpHost[]{});
    }

    @Bean
    public RestHighLevelClient restHighLevelClient() {
        builder = RestClient.builder(esHttpHost());
        setConnectTimeOutConfig();
        setMutiConnectConfig();
        restClient = builder.build();
        restHighLevelClient = new RestHighLevelClient(builder);
        log.info("init factory");
        return restHighLevelClient;
    }


    // 配置连接时间延时
    public void setConnectTimeOutConfig() {
        builder.setRequestConfigCallback(requestConfigBuilder -> {
            requestConfigBuilder.setConnectTimeout(CONNECT_TIMEOUT_MILLIS);
            requestConfigBuilder.setSocketTimeout(SOCKET_TIMEOUT_MILLIS);
            requestConfigBuilder.setConnectionRequestTimeout(CONNECTION_REQUEST_TIMEOUT_MILLIS);
            return requestConfigBuilder;
        });
    }

    // 使用异步httpclient时设置并发连接数
    public void setMutiConnectConfig() {
        builder.setHttpClientConfigCallback(httpClientBuilder -> {
            httpClientBuilder.setMaxConnTotal(MAX_CONN_TOTAL);
            httpClientBuilder.setMaxConnPerRoute(MAX_CONN_PER_ROUTE);
            return httpClientBuilder;
        });
    }

}
